package cn.doitedu.demo7;

import cn.doitedu.beans.UserAction;
import cn.doitedu.common.utils.CommonUtils;
import cn.doitedu.common.utils.StateDescriptors;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/7
 * @Desc: 学大数据，上多易教育
 * 实现： 可以动态注入、下线、更新规则
 * 当然，前提是，这些规则，都是系统已经内置的 规则模型下的规则
 **/
@Slf4j
public class MainEntry4_Final {

    //private static final Logger log = LoggerFactory.getLogger(MainEntry.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setStateBackend(new HashMapStateBackend());


        // 构造cdc连接器，实时抓取 规则元数据表中的数据变更
        SingleOutputStreamOperator<RuleMeta> ruleMetaBeans = CommonUtils.getRuleMetaStream(env);


        // 将规则元数据流，进行广播
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaBeans.broadcast(StateDescriptors.RULEMETA_BC_STATE_DES);


        // 构建kafka source来读取用户行为数据
        DataStreamSource<String> stream = CommonUtils.getActionJsonStream(env);

        // 将source读取到的json数据，转换成javaBean数据
        DataStream<UserAction> mapped = stream.map(json -> JSON.parseObject(json, UserAction.class));

        // 按用户分区
        KeyedStream<UserAction, Long> keyedStream = mapped.keyBy(UserAction::getUser_id);


        // 将用户行为流 ， 连接   规则元数据广播流
        keyedStream
                .connect(ruleMetaBroadcastStream)
                .process(new CoreProcessFunction_Final())
                .print();


        env.execute();


    }

}
